package com.example.config;

import com.example.constant.RedisConstant;
import com.example.listener.ListenerMessage1;
import com.example.listener.ListenerMessage2;
import com.example.handle.MyErrorHandle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Vector;
import java.util.concurrent.ForkJoinPool;

/**
 * @author LiYuhang
 * @version 0.1
 * @application
 * @Date 2020/9/21 16:34
 */
@Configuration
@Component
public class RedisConsumerConfig  implements DisposableBean {

    private static final Logger log = LoggerFactory.getLogger(RedisConsumerConfig.class);

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private ListenerMessage1 streamListener1;

    @Autowired
    private ListenerMessage2 streamListener2;

    @Resource
    private ForkJoinPool forkJoinPool;


    @PostConstruct
    public  void init() throws Exception{
        /**
         * 这里必须先判空，重复创建组会报错，获取不存在的key的组也会报错
         * 所以需要先判断是否存在key，在判断是否存在组
         * 我这里只有一个组，如果需要创建多个组的话则需要改下逻辑
         */
        StreamOperations<String, String, String> streamOperations = this.redisTemplate.opsForStream();
        /**
         * 如果没有消费者组，创建消费者组
         * 如果有消费者组，打印每个消费者的具体信息
         */
        if (redisTemplate.hasKey(RedisConstant.COMMENT_KEY)) {
            StreamInfo.XInfoGroups groups = streamOperations.groups(RedisConstant.COMMENT_KEY);
            if (groups.isEmpty()) {
                creatGroup();
            } else {
                groups.stream().forEach(group -> {
                    log.info("XInfoGroups:{}",group);
                    StreamInfo.XInfoConsumers consumers = streamOperations.consumers(RedisConstant.COMMENT_KEY,group.groupName());
                    log.info("XInfoConsumers:{}",consumers);
                });
            }
        } else {
            creatGroup();
        }
    }


    private StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options = null;


    private StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer;

    private Vector<StreamMessageListenerContainer<String, MapRecord<String, String, String>>> containerList = new Vector<>();

    /**
     * 创建消息
     * @param factory
     * @return
     */
    @Bean
    public Subscription subscription1(RedisConnectionFactory factory) {

        return getSubscription(factory,streamListener1,RedisConstant.COMMENT_KEY,RedisConstant.COMMENT_GROUP,RedisConstant.COMMENT_GROUP_CONSUMER_1);
    }

    @Bean
    public Subscription subscription2(RedisConnectionFactory factory) {

        return getSubscription(factory,streamListener2,RedisConstant.COMMENT_KEY,RedisConstant.COMMENT_GROUP,RedisConstant.COMMENT_GROUP_CONSUMER_2);

    }


    private Subscription getSubscription(RedisConnectionFactory factory,StreamListener streamListener,String streamKey,String group,String consumerName) {
        if (options == null) {
            options = StreamMessageListenerContainer
                    .StreamMessageListenerContainerOptions
                    .builder()
                    .batchSize(10)
                    .errorHandler(new MyErrorHandle())
                    .executor(forkJoinPool)
                    .serializer(new StringRedisSerializer())
                    .pollTimeout(Duration.ofSeconds(5))
                    .build();
        }
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = StreamMessageListenerContainer
                .create(factory,options);
        StreamOffset<String> streamOffset = StreamOffset.create(streamKey,ReadOffset.lastConsumed());

        Consumer consumer = Consumer.from(group,consumerName);

        Subscription subscription = listenerContainer.receive(consumer,streamOffset,streamListener);
        listenerContainer.start();
        this.containerList.add(listenerContainer);
        return subscription;
    }


    @Override
    public void destroy() throws Exception {
        this.containerList.forEach(StreamMessageListenerContainer::stop);
    }


    private void creatGroup() {
        StreamOperations<String, String, String> streamOperations = this.redisTemplate.opsForStream();
        String group = streamOperations.createGroup(RedisConstant.COMMENT_KEY,RedisConstant.COMMENT_GROUP);
        log.info("creatGroup:{}",group);
    }
}
